/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/server/located_block.h>
#include <kmhdfs/server/located_blocks.h>

#include <algorithm>
#include <cassert>
#include <iostream>

namespace Hdfs {
namespace Internal {

const LocatedBlock * LocatedBlocksImpl::findBlock(int64_t position) {
    int32_t targetBlockIdx = 0;
    return findBlock(position, targetBlockIdx);
}

const LocatedBlock * LocatedBlocksImpl::findBlock(int64_t position, int32_t & targetBlockIdx) {
    if (position < fileLength) {
        LocatedBlock target(position);
        std::vector<LocatedBlock>::iterator bound;

        if (blocks.empty() || position < blocks.begin()->getOffset()) {
            targetBlockIdx = 0;
            return NULL;
        }

        /*
         * bound is first block which start offset is larger than
         * or equal to position
         */
        bound = std::lower_bound(blocks.begin(), blocks.end(), target,
                                 std::less<LocatedBlock>());
        assert(bound == blocks.end() || bound->getOffset() >= position);
        LocatedBlock * retval = NULL;

        targetBlockIdx = (int32_t)(bound - blocks.begin());
        if (bound == blocks.end()) {
            retval = &blocks.back();
        } else if (bound->getOffset() > position) {
            assert(bound != blocks.begin());
            --bound;
            retval = &(*bound);
        } else {
            retval = &(*bound);
        }

        if (position < retval->getOffset()
                || position >= retval->getOffset() + retval->getNumBytes()) {
            return NULL;
        }

        return retval;
    } else {
        targetBlockIdx = (int32_t)blocks.size();
        return lastBlock.get();
    }
}

void LocatedBlocksImpl::insertRange(int32_t blockIdx, std::vector<LocatedBlock> & newBlocks) {
    int32_t oldIdx = blockIdx;
    int32_t insStart = 0, insEnd = 0;
    for(int32_t newIdx = 0; newIdx < newBlocks.size() && oldIdx < blocks.size(); newIdx++) {
        int64_t newOff = newBlocks[newIdx].getOffset();
        int64_t oldOff = blocks[oldIdx].getOffset();
        if(newOff < oldOff) {
            insEnd++;
        } else if(newOff == oldOff) {
            // replace old cached block by the new one
            blocks[oldIdx] = newBlocks[newIdx];
            if(insStart < insEnd) { // insert new blocks
                addAll(blocks, oldIdx, newBlocks, insStart, insEnd);
                oldIdx += insEnd - insStart;
            }
            insStart = insEnd = newIdx+1;
            oldIdx++;
        } else {  // newOff > oldOff
            assert(false);
        }
    }
    insEnd = (int32_t)newBlocks.size();
    if(insStart < insEnd) { // insert new blocks
        addAll(blocks, oldIdx, newBlocks, insStart, insEnd);
    }
}

void LocatedBlocksImpl::addAll(std::vector<LocatedBlock> & oldBlocks, int32_t index, std::vector<LocatedBlock> & newBlocks,
            int32_t start, int32_t end) {
    int32_t oldSize = (int32_t)oldBlocks.size();
    int32_t shiftSize = end - start;
    oldBlocks.resize(oldSize + shiftSize);
    for (int32_t i = oldSize - 1; i >= index; --i) {
        oldBlocks[i + shiftSize] = oldBlocks[i];
    }
    for (int32_t i = 0; i < shiftSize; ++i) {
        oldBlocks[index + i] = newBlocks[start + i];
    }
}

}
}
